package com.bff.gaia.client.combine.planutil;

import com.bff.gaia.api.common.Plan;
import com.bff.gaia.api.common.io.RichInputFormat;
import com.bff.gaia.api.common.io.RichOutputFormat;
import com.bff.gaia.api.common.operators.GenericDataSinkBase;
import com.bff.gaia.api.common.operators.GenericDataSourceBase;
import com.bff.gaia.api.common.operators.Operator;
import com.bff.gaia.api.common.operators.base.*;

import java.util.ArrayList;
import java.util.List;

public class PlanDataFlowInformation {
	// 一个用户提交的plan
	private Plan plan;
	// plan类型条件 控制迭代和输出类型
	private boolean isIteType = true;  // 用来判段一个Plan中是否有迭代算子,有则为false
	private boolean pritnType = true;   // 判断sink中是否存在print类型的 输出类型 如果存在设置为 false
	// 每个plan的属性信息
	private int operatorSumNumber = 0;//Plan中总的算子个数
	private int avLength = 0;
	private int[][] operatorNumber = new int[19][2];
	private int dept = 0;
	private int sourceParallizer = -1;

	private int maxParallizer = 0; // plan中所有算子中并行度最大的值
	private int pParallizer = 0; // Plan defaultParallelism 并行度

	private int workComplex = 0;

	//算子所在位置 指向上面的 operatorNumber

	private final int genericDataSinkBase = 0;
	private final int bulkIterationBase = 1;
	private final int deltaIterationBase = 2;

	private final int coGroupOperatorBase = 3;
	private final int coGroupRawOperatorBase = 4;

	private final int crossOperatorBase = 5;

	private final int filterOperatorBase = 6;
	private final int flatMapOperatorBase = 7;
	private final int mapOperatorBase = 8;

	private final int groupCombineOperatorBase = 10;

	private final int groupReduceOperatorBase = 11;
	private final int reduceOperatorBase = 12;

	private final int innerJoinOperatorBase = 13;
	private final int joinOperatorBase = 14;
	private final int outerJoinOperatorBase = 15;

	private final int mapPartitionOperatorBase = 9;
	private final int partitionOperatorBase = 16;
	private final int sortPartitionOperatorBase = 17;

	private final int genericDataSourceBase = 18;

	private String[] suanzi = {"GenericDataSinkBase ", "BulkIterationBase ", "DeltaIterationBase ", "CoGroupOperatorBase ", "CoGroupRawOperatorBase ", "CrossOperatorBase ", "FilterOperatorBase ", "FlatMapOperatorBase ", "MapOperatorBase ", "MapPartitionOperatorBase ", "GroupCombineOperatorBase ", "GroupReduceOperatorBase ", "ReduceOperatorBase ", "InnerJoinOperatorBase ", "JoinOperatorBase ", "OuterJoinOperatorBase ", "PartitionOperatorBase ", "SortPartitionOperatorBase ", "GenericDataSourceBase "};

	private List<String> sinkFiles = new ArrayList<>();
	private List<String> sourceFiles = new ArrayList<>();

	public PlanDataFlowInformation(Plan plan) {// 初始化Plan信息
		this.plan = plan;
		this.pParallizer = plan.getDefaultParallelism();
		for (int i = 0; i < 19; i++)
			for (int j = 0; j < 2; j++)
				operatorNumber[i][j] = 0;
		getPlanInformation();//获取Plan的信息
	}

	//ptintType 为false表示有的文件输出为print类型，这样类型的任务不参与合并过程
	public boolean isPritnType() {
		return pritnType;
	}

	public int[][] getOperatorNumbersAndParallizer() { //获取每个算子个数的详细情况
		return operatorNumber;
	}

	public int getOperatorSumNumber() {//获取总共的算子数
		return operatorSumNumber;
	}

	public List getSinkFiles() {//获取每个输出文件路径
		return sinkFiles;
	}

	public List getSourceFiles() {//获取每个输入文件的路径
		return sourceFiles;
	}

	public boolean planType() {//获取Plan是不是迭代类型
		return isIteType;
	}

	public int getAvLength() {//获取平均链的长度
		return avLength;
	}

	public int getDept() {//获取Plan中任务链的最大长度
		return dept;
	}

	public String[] getSuanzi() {
		return suanzi;
	}

	public Plan getPlan() {// 返回 plan
		return plan;
	}

	private void setMaxParallizer() {//获取算子最大的并行度
		int max = -1;
		for (int i = 0; i < 19; i++) {
			if (operatorNumber[i][1] > max)
				max = operatorNumber[i][1];
		}
		maxParallizer = max;
	}

	public int getMaxParallizer() {
		setMaxParallizer();
		return maxParallizer;
	}

	private void setSourceParallizer() {//获取源节点最大并行度
		sourceParallizer = operatorNumber[18][1];
	}

	public int getSourceParallizer() {
		setSourceParallizer();
		return sourceParallizer;
	}


	private void getPlanInformation() {//获取Plan中各种信息
		List<GenericDataSinkBase<?>> sinks = (List<GenericDataSinkBase<?>>) plan.getDataSinks();
		int max = 0, med = 0;
		//调用 getDeptOfPlan 方法获取作业执行图的最大深度
		for (GenericDataSinkBase sink : sinks) {
			med = getDeptOfPlan(sink, sink);
			if (med > max)
				max = med;
		}
		dept = max;

		// 用于检测 文件的 类型是否出错
		try {
			// 调用 setPlanElemsOutputValues
			for (GenericDataSinkBase sink : sinks)
				setPlanElemsOutputValues(sink, sink);
		} catch (NullPointerException npe) {
			System.out.println("setPlanElemsOutputValues 方法出现空指针错误： NullPointerException");
			pritnType = false;
			return;
		}
		avLength = operatorSumNumber / sourceFiles.size();
	}

	// 遍历plan中的算子得到任务的部分信息，算子总的个数，每种类型算子的个数
	public void setPlanElemsOutputValues(Operator operator, Operator output) {
		if (operator == null) {
			return;
		}
		operatorSumNumber++;  // 计算图中存在多少的节点
		if (operator instanceof GenericDataSinkBase) { // 1

			operatorNumber[genericDataSinkBase][0]++;

			GenericDataSinkBase gensink = (GenericDataSinkBase) operator;

			if (gensink.getParallelism() > operatorNumber[genericDataSinkBase][1])
				operatorNumber[genericDataSinkBase][1] = gensink.getParallelism();
			RichOutputFormat outputFormat = (RichOutputFormat) gensink.getFormatWrapper().getUserCodeObject();
			/**
			 * test Path if is null
			 */
			if(outputFormat==null){
				System.out.println("当前节点的输出为空");
			}
			else{
				if(outputFormat.getOutputFilePath()==null){
					System.out.println("作业的输出文件是空");
				}
				else{
					String path=outputFormat.getOutputFilePath().getPath();
					System.out.println("文件的路径不为空："+path);
					sinkFiles.add(outputFormat.getOutputFilePath().getPath());
				}

			}
			if(gensink.getInput()!=null){
				setPlanElemsOutputValues(gensink.getInput(), gensink);
			}
			else{
				System.out.println("当前作业节点的前驱为空！");
			}

		} else if (operator instanceof GroupReduceOperatorBase) {// 2
			System.out.println("是groupReduce算子");
			operatorNumber[groupReduceOperatorBase][0]++;
			GroupReduceOperatorBase gROB = (GroupReduceOperatorBase) operator;
			if (gROB.getParallelism() > operatorNumber[groupReduceOperatorBase][1])
				operatorNumber[groupReduceOperatorBase][1] = gROB.getParallelism();
			if (!gROB.isTraved()) {
				setPlanElemsOutputValues(gROB.getInput(), gROB);
				gROB.setTraved(true);
			}
		} else if (operator instanceof FilterOperatorBase) {// 3
			System.out.println("是FilterOperatorBase算子");
			operatorNumber[filterOperatorBase][0]++;
			FilterOperatorBase fOB = (FilterOperatorBase) operator;
			if (fOB.getParallelism() > operatorNumber[filterOperatorBase][1])
				operatorNumber[filterOperatorBase][1] = fOB.getParallelism();
			if (!fOB.isTraved()) {
				setPlanElemsOutputValues(fOB.getInput(), fOB);
				fOB.setTraved(true);
			}
		} else if (operator instanceof SortPartitionOperatorBase) {//4
			System.out.println("是SortPartitionOperatorBase算子");
			operatorNumber[sortPartitionOperatorBase][0]++;

			SortPartitionOperatorBase sPOB = (SortPartitionOperatorBase) operator;

			if (sPOB.getParallelism() > operatorNumber[sortPartitionOperatorBase][1])
				operatorNumber[sortPartitionOperatorBase][1] = sPOB.getParallelism();

			if (!sPOB.isTraved()) {
				setPlanElemsOutputValues(sPOB.getInput(), sPOB);
				sPOB.setTraved(true);
			}
		} else if (operator instanceof PartitionOperatorBase) {//5
			System.out.println("是PartitionOperator算子");
			operatorNumber[partitionOperatorBase][0]++;

			PartitionOperatorBase pOB = (PartitionOperatorBase) operator;

			if (pOB.getParallelism() > operatorNumber[partitionOperatorBase][1])
				operatorNumber[partitionOperatorBase][1] = pOB.getParallelism();

			if (!pOB.isTraved()) {
				setPlanElemsOutputValues(pOB.getInput(), pOB);
				pOB.setTraved(true);
			}
		} else if (operator instanceof MapPartitionOperatorBase) { //6
			System.out.println("是MapPartitionOperator算子");
			operatorNumber[mapPartitionOperatorBase][0]++;

			MapPartitionOperatorBase mPOB = (MapPartitionOperatorBase) operator;

			if (mPOB.getParallelism() > operatorNumber[mapPartitionOperatorBase][1])
				operatorNumber[mapPartitionOperatorBase][1] = mPOB.getParallelism();

			if (mPOB.traved) {
				setPlanElemsOutputValues(mPOB.getInput(), mPOB);
				mPOB.traved = true;
			}
		} else if (operator instanceof FlatMapOperatorBase) { //7
			System.out.println("是FlatMapOperator算子");
			operatorNumber[flatMapOperatorBase][0]++;
			FlatMapOperatorBase fMOB = (FlatMapOperatorBase) operator;

			if (fMOB.getParallelism() > operatorNumber[flatMapOperatorBase][1])
				operatorNumber[flatMapOperatorBase][1] = fMOB.getParallelism();

			if (!fMOB.traved) {
				setPlanElemsOutputValues(fMOB.getInput(), fMOB);
				fMOB.traved = true;
			}
		} else if (operator instanceof MapOperatorBase) { //8
			System.out.println("是MapOperator算子");
			operatorNumber[mapOperatorBase][0]++;
			MapOperatorBase mOB = (MapOperatorBase) operator;

			if (mOB.getParallelism() > operatorNumber[mapOperatorBase][1])
				operatorNumber[mapOperatorBase][1] = mOB.getParallelism();

			if (!mOB.traved) {
				setPlanElemsOutputValues(mOB.getInput(), mOB);
				mOB.traved = true;
			}
		} else if (operator instanceof ReduceOperatorBase) {//9
			System.out.println("是ReduceOperator算子");
			operatorNumber[reduceOperatorBase][0]++;
			ReduceOperatorBase rOB = (ReduceOperatorBase) operator;
			if (rOB.getParallelism() > operatorNumber[reduceOperatorBase][1])
				operatorNumber[reduceOperatorBase][1] = rOB.getParallelism();
			if (!rOB.traved) {
				setPlanElemsOutputValues(rOB.getInput(), rOB);
				rOB.traved = true;
			}
		} else if (operator instanceof GroupCombineOperatorBase) {//10
			System.out.println("是GroupCombine算子");
			operatorNumber[groupCombineOperatorBase][0]++;

			GroupCombineOperatorBase gCOB = (GroupCombineOperatorBase) operator;

			if (gCOB.getParallelism() > operatorNumber[groupReduceOperatorBase][1])
				operatorNumber[groupReduceOperatorBase][1] = gCOB.getParallelism();

			if (!gCOB.traved) {
				setPlanElemsOutputValues(gCOB.getInput(), gCOB);
				gCOB.traved = true;
			}
		} else if (operator instanceof BulkIterationBase) { //迭代相关先不用加 11
			System.out.println("是BulkIteration算子");
			operatorNumber[bulkIterationBase][0]++;
			isIteType = false;
			return;
		} else if (operator instanceof DeltaIterationBase) { // 迭代不加 12
			System.out.println("是DeltaIteration算子");
			operatorNumber[deltaIterationBase][0]++;
			this.isIteType = false;
			return;
		} else if (operator instanceof CoGroupOperatorBase) { // 双输入节点 13
			System.out.println("是CoGroupOperator算子");
			operatorNumber[coGroupOperatorBase][0]++;

			CoGroupOperatorBase cGOB = (CoGroupOperatorBase) operator;

			if (cGOB.getParallelism() > operatorNumber[coGroupOperatorBase][1])
				operatorNumber[coGroupOperatorBase][1] = cGOB.getParallelism();

			if (!cGOB.traved) {
				setPlanElemsOutputValues(cGOB.getFirstInput(), cGOB);
				setPlanElemsOutputValues(cGOB.getSecondInput(), cGOB);
				cGOB.traved = true;
			}
		} else if (operator instanceof CrossOperatorBase) { //双输入节点 14
			System.out.println("是CrossOperator算子");
			operatorNumber[crossOperatorBase][0]++;

			CrossOperatorBase cOB = (CrossOperatorBase) operator;

			if (cOB.getParallelism() > operatorNumber[crossOperatorBase][1])
				operatorNumber[crossOperatorBase][1] = cOB.getParallelism();

			if (!cOB.traved) {
				setPlanElemsOutputValues(cOB.getFirstInput(), cOB);
				setPlanElemsOutputValues(cOB.getSecondInput(), cOB);
				cOB.traved = true;
			}
		} else if (operator instanceof CoGroupRawOperatorBase) { //双输入节点 15
			System.out.println("是CoGroupRawOperator算子");
			operatorNumber[coGroupRawOperatorBase][0]++;

			CoGroupRawOperatorBase cGROB = (CoGroupRawOperatorBase) operator;

			if (cGROB.getParallelism() > operatorNumber[coGroupRawOperatorBase][1])
				operatorNumber[coGroupRawOperatorBase][1] = cGROB.getParallelism();

			if (!cGROB.traved) {
				setPlanElemsOutputValues(cGROB.getFirstInput(), cGROB);
				setPlanElemsOutputValues(cGROB.getSecondInput(), cGROB);
				cGROB.traved = true;
			}
		} else if (operator instanceof JoinOperatorBase) {//双输入节点 16
			System.out.println("是JoinOperator算子");
			operatorNumber[joinOperatorBase][0]++;

			JoinOperatorBase jOB = (JoinOperatorBase) operator;

			if (jOB.getParallelism() > operatorNumber[joinOperatorBase][1])
				operatorNumber[joinOperatorBase][1] = jOB.getParallelism();

			if (!jOB.traved) {
				setPlanElemsOutputValues(jOB.getFirstInput(), jOB);
				setPlanElemsOutputValues(jOB.getSecondInput(), jOB);
				jOB.traved = true;
			}
		} else if (operator instanceof InnerJoinOperatorBase) {//双输入节点 17

			operatorNumber[innerJoinOperatorBase][0]++;

			InnerJoinOperatorBase iJOB = (InnerJoinOperatorBase) operator;

			if (iJOB.getParallelism() > operatorNumber[innerJoinOperatorBase][1])
				operatorNumber[innerJoinOperatorBase][1] = iJOB.getParallelism();
			if (!iJOB.traved) {
				setPlanElemsOutputValues(iJOB.getFirstInput(), iJOB);
				setPlanElemsOutputValues(iJOB.getSecondInput(), iJOB);
				iJOB.traved = true;
			}
		} else if (operator instanceof OuterJoinOperatorBase) {//双输入节点 18

			operatorNumber[outerJoinOperatorBase][0]++;

			OuterJoinOperatorBase oJOB = (OuterJoinOperatorBase) operator;

			if (oJOB.getParallelism() > operatorNumber[outerJoinOperatorBase][1])
				operatorNumber[outerJoinOperatorBase][1] = oJOB.getParallelism();

			if (!oJOB.traved) {
				setPlanElemsOutputValues(oJOB.getFirstInput(), oJOB);
				setPlanElemsOutputValues(oJOB.getSecondInput(), oJOB);
				oJOB.traved = true;
			}
		} else if (operator instanceof GenericDataSourceBase) { // source 节点 19
			System.out.println("输入文件源是DataSource节点！！！！！！");
			operatorNumber[genericDataSourceBase][0]++;

			GenericDataSourceBase gDSB = (GenericDataSourceBase) operator;

			if (gDSB.getParallelism() > operatorNumber[genericDataSourceBase][1])
				operatorNumber[genericDataSourceBase][1] = gDSB.getParallelism();

			RichInputFormat input = (RichInputFormat) gDSB.getFormatWrapper().getUserCodeObject();
			if(input.getFilePath().getPath()==null){
				System.out.println("文件的输入路径是空-----》");
			}
			sourceFiles.add(input.getFilePath().getPath().toString());

			System.out.println("SourceFile Size:" + sourceFiles.size());
			if (!gDSB.isTraved()) {
				gDSB.setTraved(true);
			}
		}
	}

	public int getDeptOfPlan(Operator operator, Operator output) { // 获取任务链的最大长度
		if (operator == null) {
			return 0;
		}
		int branch_1 = 0;
		int branch_2 = 0;
		if (operator instanceof GenericDataSinkBase) { // 1
			GenericDataSinkBase gensink = (GenericDataSinkBase) operator;
			branch_1 = getDeptOfPlan(gensink.getInput(), gensink);
			return branch_1 + 1;
		} else if (operator instanceof GroupReduceOperatorBase) {// 2
			GroupReduceOperatorBase groupReduceOperatorBase = (GroupReduceOperatorBase) operator;
			branch_1 = getDeptOfPlan(groupReduceOperatorBase.getInput(), groupReduceOperatorBase);
			return branch_1 + 1;
		} else if (operator instanceof FilterOperatorBase) {// 3
			FilterOperatorBase filterOperatorBase = (FilterOperatorBase) operator;
			branch_1 = getDeptOfPlan(filterOperatorBase.getInput(), filterOperatorBase);
			return branch_1 + 1;
		} else if (operator instanceof SortPartitionOperatorBase) {//4
			SortPartitionOperatorBase sortPartitionOperatorBase = (SortPartitionOperatorBase) operator;
			branch_1 = getDeptOfPlan(sortPartitionOperatorBase.getInput(), sortPartitionOperatorBase);
			return branch_1 + 1;
		} else if (operator instanceof PartitionOperatorBase) {//5
			PartitionOperatorBase partitionOperatorBase = (PartitionOperatorBase) operator;
			branch_1 = getDeptOfPlan(partitionOperatorBase.getInput(), partitionOperatorBase);
			return branch_1 + 1;
		} else if (operator instanceof MapPartitionOperatorBase) { //6
			MapPartitionOperatorBase mapPartitionOperatorBase = (MapPartitionOperatorBase) operator;
			branch_1 = getDeptOfPlan(mapPartitionOperatorBase.getInput(), mapPartitionOperatorBase);
			return branch_1 + 1;
		} else if (operator instanceof FlatMapOperatorBase) { //7
			FlatMapOperatorBase flatMapOperatorBase = (FlatMapOperatorBase) operator;
			branch_1 = getDeptOfPlan(flatMapOperatorBase.getInput(), flatMapOperatorBase);
			return branch_1 + 1;
		} else if (operator instanceof MapOperatorBase) { //8
			MapOperatorBase mapOperatorBase = (MapOperatorBase) operator;
			branch_1 = getDeptOfPlan(mapOperatorBase.getInput(), mapOperatorBase);
			return branch_1 + 1;
		} else if (operator instanceof ReduceOperatorBase) {//9
			ReduceOperatorBase reduceOperatorBase = (ReduceOperatorBase) operator;
			branch_1 = getDeptOfPlan(reduceOperatorBase.getInput(), reduceOperatorBase);
			return branch_1 + 1;
		} else if (operator instanceof GroupCombineOperatorBase) {//10
			GroupCombineOperatorBase groupCombineOperatorBase = (GroupCombineOperatorBase) operator;
			branch_1 = getDeptOfPlan(groupCombineOperatorBase.getInput(), groupCombineOperatorBase);
			return branch_1 + 1;
		} else if (operator instanceof BulkIterationBase) { //迭代相关先不用加 11
			this.isIteType = false;
			return 0;
		} else if (operator instanceof DeltaIterationBase) { // 迭代不加 12
			this.isIteType = false;
			return 0;
		} else if (operator instanceof CoGroupOperatorBase) { // 双输入节点 13
			CoGroupOperatorBase coGroupOperatorBase = (CoGroupOperatorBase) operator;
			branch_1 = getDeptOfPlan(coGroupOperatorBase.getFirstInput(), coGroupOperatorBase) + 1;
			branch_2 = getDeptOfPlan(coGroupOperatorBase.getSecondInput(), coGroupOperatorBase) + 1;
			return branch_1 > branch_2 ? branch_1 : branch_2;
		} else if (operator instanceof CrossOperatorBase) { //双输入节点 14
			CrossOperatorBase crossOperatorBase = (CrossOperatorBase) operator;
			branch_1 = getDeptOfPlan(crossOperatorBase.getFirstInput(), crossOperatorBase) + 1;
			branch_2 = getDeptOfPlan(crossOperatorBase.getSecondInput(), crossOperatorBase) + 1;
			return branch_1 > branch_2 ? branch_1 : branch_2;
		} else if (operator instanceof CoGroupRawOperatorBase) { //双输入节点 15
			CoGroupRawOperatorBase coGroupRawOperatorBase = (CoGroupRawOperatorBase) operator;
			branch_1 = getDeptOfPlan(coGroupRawOperatorBase.getFirstInput(), coGroupRawOperatorBase) + 1;
			branch_2 = getDeptOfPlan(coGroupRawOperatorBase.getSecondInput(), coGroupRawOperatorBase) + 1;
			return branch_1 > branch_2 ? branch_1 : branch_2;
		} else if (operator instanceof JoinOperatorBase) {//双输入节点 16
			JoinOperatorBase joinOperatorBase = (JoinOperatorBase) operator;
			branch_1 = getDeptOfPlan(joinOperatorBase.getFirstInput(), joinOperatorBase) + 1;
			branch_2 = getDeptOfPlan(joinOperatorBase.getSecondInput(), joinOperatorBase) + 1;
			return branch_1 > branch_2 ? branch_1 : branch_2;
		} else if (operator instanceof InnerJoinOperatorBase) {//双输入节点 17
			InnerJoinOperatorBase innerJoinOperatorBase = (InnerJoinOperatorBase) operator;
			branch_1 = getDeptOfPlan(innerJoinOperatorBase.getFirstInput(), innerJoinOperatorBase) + 1;
			branch_2 = getDeptOfPlan(innerJoinOperatorBase.getSecondInput(), innerJoinOperatorBase) + 1;
			return branch_1 > branch_2 ? branch_1 : branch_2;
		} else if (operator instanceof OuterJoinOperatorBase) {//双输入节点 18
			OuterJoinOperatorBase outerJoinOperatorBase = (OuterJoinOperatorBase) operator;
			branch_1 = getDeptOfPlan(outerJoinOperatorBase.getFirstInput(), outerJoinOperatorBase) + 1;
			branch_2 = getDeptOfPlan(outerJoinOperatorBase.getSecondInput(), outerJoinOperatorBase) + 1;
			return branch_1 > branch_2 ? branch_1 : branch_2;
		} else if (operator instanceof GenericDataSourceBase) { // source 节点 19
			GenericDataSourceBase genericDataSourceBase = (GenericDataSourceBase) operator;
			return 1;
		}
		return 0;
	}

	public int getParallizer() { //获取任务的并行度
           if(pParallizer>maxParallizer){
           	return pParallizer;
		   }else {
           	return maxParallizer;
		   }
	}

	private void setWorkComplex() {	// 通过计算算子中 reduce和join类算子的总的个数来计算任务的复杂度
		workComplex = operatorNumber[joinOperatorBase][0]
			+ operatorNumber[outerJoinOperatorBase][0]
			+ operatorNumber[innerJoinOperatorBase][0]
			+ operatorNumber[crossOperatorBase][0]
			+ operatorNumber[reduceOperatorBase][0]
			+ operatorNumber[groupReduceOperatorBase][0];
	}
	public int getWorkComplex()
	{
		setWorkComplex();
		return workComplex;
	}
	public boolean getType(){
		return isIteType && pritnType;
	}
}